package com.kdpujie;

import java.util.Random;

import org.apache.flink.streaming.api.functions.source.SourceFunction;

/**
 * 模拟生成持续的event
 */
public class EventSource implements SourceFunction<UserEvent>{
    private boolean isRunning = true;
    private Random rand = new Random();
    private Long inc = 1l;

    @Override
    public void cancel() {
        this.isRunning = false;
    }

    @Override
    public void run(SourceContext<UserEvent> ctx) throws Exception {
        while(this.isRunning) {
            ctx.collect(this.next());
        }
    }

    // 生成下一个用户事件。事件间隔：0-1000ms
    private UserEvent next() {
        try {
            Thread.sleep(rand.nextInt(1000));
        } catch (Exception e) {
            
        }
        UserEvent ue = new UserEvent();
        ue.setRequestId(inc);
        inc++;
        ue.setUserId(rand.nextInt(10) + 1);
        ue.setItermId(rand.nextInt(100) + 1);
        ue.setTs(System.currentTimeMillis());
        int tmp = rand.nextInt(100);
        if (tmp / 5 == 0) {
            ue.setAction("click");
        } else {
            ue.setAction("impression");
        }
        return ue;
    }
    
}